package com.shujia.tour

import java.text.SimpleDateFormat
import java.util.Date

import com.shujia.common.util.{Geography, SparkTool}
import org.apache.spark.sql._

object CityTouristJob extends SparkTool {
  def run(spark: SparkSession): Unit = {

    /**
      * 计算市游客
      *
      * 1、用户在城市中停留时间大于3小时
      * 2、用户出游距离大于10km
      *
      */

    import spark.implicits._
    import org.apache.spark.sql.functions._


    //加载停留表
    val stayPoint: Dataset[Row] = spark
      .table("dwi.dwi_staypoint_msk_d")
      .where($"day_id" === day_id)


    //用户 画像表
    val usertag: Dataset[Row] = spark.table("dim.dim_usertag_msk_m")
      .where($"month_id" === month_id)


    //行政区配置表
    val adminCode: Dataset[Row] = spark.table("dim.dim_admincode")


    //计算两个网格点距离的函数
    spark
      .udf
      .register(
        "calculateLength",
        (grid1: String, grid2: String) => Geography.calculateLength(grid1.toLong, grid2.toLong)
      )


    val cityTour: DataFrame = stayPoint
      .join(adminCode.hint("broadcast"), "county_id") //管理行政区配置表
      .select($"mdn", $"city_id", $"duration", $"grid_id")
      .join(usertag, "mdn") //关联用户画像表
      //计算每隔点到常住地的距离
      .select($"mdn", $"city_id", $"duration", expr("calculateLength(grid_id,resi_grid_id) / 1000.0 as distance"), $"resi_county_id")
      .groupBy($"mdn", $"city_id", $"resi_county_id") //按手机号和城市编号分组
      .agg(sum($"duration") / 60.0 as "d_stay_time", max($"distance") as "d_max_distance")
      //过滤游客
      .where($"d_stay_time" > 3 and $"d_max_distance" > 10)
      .select($"mdn", $"resi_county_id" as "source_county_id", $"city_id" as "d_city_id", round($"d_stay_time", 3), round($"d_max_distance", 3))


    //打印物理计划
    cityTour.explain()


    cityTour.write
      .format("csv")
      .option("sep", "\t")
      .mode(SaveMode.Overwrite)
      .save(s"/daas/motl/dal_tour/dal_tour_city_tourist_msk_d/day_id=$day_id")

  }
}
